package panda.planner.impl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;

import panda.metainfo.IndexToken;
import panda.planner.QueryPlanner;
import panda.query.aggr.AggregateFn;
import panda.query.aggr.CountFn;
import panda.query.aggr.IntAvgFn;
import panda.query.aggr.IntSumFn;
import panda.query.aggr.MaxFn;
import panda.query.aggr.MinFn;
import panda.query.condition.Clause;
import panda.query.condition.ConditionExtract;
import panda.query.condition.Relation;
import panda.query.condition.Sentence;
import panda.query.struct.Attribute;
import panda.query.struct.Constant;
import panda.query.struct.SemAttribute;
import panda.query.tree.DistinctNode;
import panda.query.tree.GroupByNode;
import panda.query.tree.MergeJoinNode;
import panda.query.tree.OrderByNode;
import panda.query.tree.ProductNode;
import panda.query.tree.ProjectNode;
import panda.query.tree.QueryTreeNode;
import panda.query.tree.ResultSetNode;
import panda.query.tree.SelectNode;
import panda.query.tree.SortNode;
import panda.query.tree.TableNode;
import panda.query.tree.index.IndexJoinNode;
import panda.query.tree.index.IndexSelectNode;
import panda.server.Panda;
import panda.syntax.AttributeValue;
import panda.syntax.SelectStm;
import panda.syntax.TblRef;
import panda.syntax.SelectStm.AggAttr;
import panda.transaction.Transaction;

/**
 * A query planner that chooses an optimal query plan utilizing
 * dynamic programming
 * 
 * @author Tian Yuan
 *
 */
public class DPQueryPlanner implements QueryPlanner {

	private int pw[] = null;

	@Override
	public QueryTreeNode createQueryTree(SelectStm stm, Transaction tx) {
		Collection<TblRef> tbl = stm.getTables();
		Sentence cond = stm.getWhereCondition();
		Map<Integer, Element> prev, current;
		Map<TblRef, Integer> tblID = new HashMap<TblRef, Integer>();
		ArrayList<Map<String, IndexToken>> indices = new ArrayList<Map<String, IndexToken>>();

		prev = new HashMap<Integer, Element>();
		setPower(tbl.size());
		int cnt = 0;
		for (TblRef r : tbl) {
			tblID.put(r, cnt);

			Map<String, IndexToken> ti = Panda.getMetaInfoManager().getAllIndices(r.getTableName(), tx);

			indices.add(ti);

			TableNode tn = new TableNode(r.getToken(), tx);

			//			System.out.println("HELLO!");
			//			for (Attribute a : r.getToken().getSchema().getAllAttributes())
			//				System.out.println(a);
			//			System.out.println("NONONO");

			prev.put(pw[cnt], getProperSelect(stm, new Element(tn, new HashSet<Clause>()) , cond, ti, tx));
			cnt ++;
		}

		while (prev.size() > 1) {
			current = new HashMap<Integer, Element>();
			Collection<Entry<Integer, Element>> pSet = prev.entrySet();

			for (TblRef r : tbl) {
				int id = tblID.get(r); 

				for (Entry<Integer, Element> i : pSet) {
					//					System.out.println(i.getKey() + " , " + id);
					//					System.out.println(i.getValue().node.getSchema().getAttributeByName("a") + "," + r.getToken().getSchema().getAttributeByName("a"));
					if ((i.getKey() & pw[id]) == 0){
						Element newVal = getMinPlan(stm, i.getValue(), r, cond, indices.get(id), tx);
						Element orgMin = current.get(i.getKey() + pw[id]);

						if (orgMin == null || orgMin.getCost() > newVal.getCost())
							current.put(i.getKey() + pw[id], newVal);
					}
				}
			}
			prev = current;
		}

		Element e = prev.get(pw[cnt] - 1);
		
		QueryTreeNode res = e.node;
		
		Sentence rem = ConditionExtract.getRemaining(stm.getWhereCondition(), e.used);
		if (rem != null) res = new SelectNode(res, rem, tx);

		if (stm.groupbySem.size() != 0 || stm.aggSem.size() != 0) {
			List<AggregateFn> agg = new ArrayList<AggregateFn>();

			for (AggAttr aa : stm.aggSem) {
				switch (aa.func) {
				case AttributeValue.AVG : agg.add(new IntAvgFn(aa.bound, aa.newattr));break;
				case AttributeValue.MAX : agg.add(new MaxFn(aa.bound, aa.newattr));break;
				case AttributeValue.MIN : agg.add(new MinFn(aa.bound, aa.newattr));break;
				case AttributeValue.SUM : agg.add(new IntSumFn(aa.bound, aa.newattr));break;
				case AttributeValue.COUNT : agg.add(new CountFn(aa.bound, aa.newattr));break;
				}
			}

			res = new GroupByNode(res, stm.groupbySem, agg, tx);

			Sentence havingCon = stm.getHavingCondition();
			if (havingCon != null)
				res = new SelectNode(res, havingCon, tx);
		}

		if (stm.orderbySem.size() != 0) {

			if (stm.distinct) {

				ArrayList<AggregateFn> list = new ArrayList<AggregateFn>();
				ArrayList<Attribute> tattr = new ArrayList<Attribute>();
				ArrayList<Boolean> tord = new ArrayList<Boolean>();

				for (int i = 0; i < stm.orderbySem.size(); i++) {
					Attribute a = stm.orderbySem.get(i);
					if (!stm.selectAttr.contains(a)) {
						if (stm.ascOrder.get(i))
							list.add(new MinFn(a, new SemAttribute("MIN(" + a.getAttributeName() + ")")));
						else
							list.add(new MaxFn(a, new SemAttribute("MAX(" + a.getAttributeName() + ")")));

						tattr.add(list.get(list.size() - 1).getAttribute());
					}
					else
						tattr.add(a);
					
					tord.add(stm.ascOrder.get(i));

				}

				for (int i = 0; i < stm.selectAttr.size(); i++) {
					Attribute a = stm.selectAttr.get(i);
					
					if (!stm.orderbySem.contains(a)) {
						tattr.add(a);
						tord.add(true);
					}
				}

				if (list.size() != 0)
					res = new GroupByNode(res, stm.selectAttr, list, tx);

				res = new OrderByNode(res, tattr, tord, tx);
				res = new DistinctNode(res, stm.selectAttr, tx);
			}
			else
				res = new OrderByNode(res, stm.orderbySem, stm.ascOrder, tx);
		}
		else
			if (stm.distinct) {
				res = new SortNode(res, stm.selectAttr, tx);
				res = new DistinctNode(res, stm.selectAttr, tx);
			}

		return new ResultSetNode(res, stm.attrs, stm.star, tx);
	}

	/**
	 * Pre-calculate the power of 2 into an array 
	 * @param size the maximum size of the result
	 */
	private void setPower(int size) {
		if (pw == null || size > pw[pw.length - 1]) {
			pw = new int[(int)Math.ceil(Math.log(size) / Math.log(2)) + 2];

			int tmp = 1;
			for (int i = 0; i < pw.length; i++) {
				pw[i] = tmp;
				tmp *= 2;
			}
		}
	}

	/**
	 * Get the optimal plan that joins two tables
	 * @param org
	 * @param tbl
	 * @return
	 */
	private Element getMinPlan(SelectStm stm, Element org, TblRef tbl, Sentence s, Map<String, IndexToken> indices, Transaction tx) {
		TableNode tn = new TableNode(tbl.getToken(), tx);
		Element te = getProperSelect(stm, new Element(tn, new HashSet<Clause>()), s, indices, tx);
		HashSet<Clause> ts1 = new HashSet<Clause>(org.used);

		if (s != null) {
			ts1.addAll(te.used);
			Sentence cond = ConditionExtract.getJoinCondition(org.node.getSchema(), te.node.getSchema(), ts1, s);


			if (cond != null) {
				HashSet<Clause> ts2 = new HashSet<Clause>(ts1);
				Element ta = null, tb;

				Attribute[] joinAttr = ConditionExtract.getFirstJoinIndex(te.node.getSchema(), cond, indices, ts1);

				if (joinAttr != null)
					ta = getProperSelect(stm, new Element(new IndexJoinNode(org.node, te.node, indices.get(joinAttr[1].getAttributeName()), joinAttr[0], tx), ts1), cond, indices, tx);

				List<Attribute>[] col = ConditionExtract.covertToColumn(org.node.getSchema(), te.node.getSchema(), cond);
				tb = getProperSelect(stm, new Element(new MergeJoinNode(org.node, te.node, col[0], col[1], tx), ts2), cond, indices, tx);

				if (ta != null && ta.getCost() < tb.getCost())
					return ta;
				else
					return tb;
			}
		}
		return getProperSelect(stm, new Element(new ProductNode(org.node, te.node, tx), ts1), s, indices, tx);
	}

	/**
	 * Generate a select node that can be adapted to the current tree
	 * @param n the tree to which the select node is put
	 * @param used the clauses that are already applied to the current tree
	 * @return the current tree added by a select node
	 */
	private Element getProperSelect(SelectStm stm, Element n, Sentence s, Map<String, IndexToken> map, Transaction tx) {
		if (s == null) {
			HashSet<Clause> ts1 = new HashSet<Clause>(n.used);
			return new Element(getProperProject(stm, n.node, ConditionExtract.getProject(s, ts1), tx), ts1);
		}

		HashSet<Clause> ts1 = new HashSet<Clause>(n.used), ts2 = new HashSet<Clause>(n.used); 



		Sentence cond = ConditionExtract.getSelectCondition(n.node.getSchema(), ts1, s);
		if (cond == null) {
			HashSet<Clause> hs = new HashSet<Clause>(n.used);
			return new Element(getProperProject(stm, n.node, ConditionExtract.getProject(s, hs), tx), hs);
		}

		QueryTreeNode qn = new SelectNode(n.node, cond, tx);
		Element n1 = new Element(getProperProject(stm, qn, ConditionExtract.getProject(s, ts1), tx), ts1);
		Element n2;

		if (map == null)
			n2 = null;
		else {
			Relation indCond = ConditionExtract.getOptSelectIndex(n.node, cond, map, ts2);
			Attribute attr = ConditionExtract.getAttribute(indCond);
			Constant con = ConditionExtract.getConstant(indCond);
			int op = ConditionExtract.getOp(indCond);

			if (indCond != null) {
				n2 = getProperSelect(stm, new Element(new IndexSelectNode(n.node, map.get(attr.getAttributeName()), con, op, tx), ts2), cond, map, tx);
				n2 = new Element(getProperProject(stm, n2.node, ConditionExtract.getProject(s, ts2), tx), ts2);
			}
			else
				n2 = null;
		}

		if (n2 == null || n1.getCost() < n2.getCost())
			return n1;
		else
			return n2;
	}

	private QueryTreeNode getProperProject(SelectStm stm, QueryTreeNode qn, Set<Attribute> proj, Transaction tx) {
		//		return qn;
		Collection<Attribute> attr = new HashSet<Attribute>(proj);
		Collection<Attribute> pa = new ArrayList<Attribute>();
		attr.addAll(stm.selectAttr);
		attr.addAll(stm.fixedAttr);
		attr.addAll(stm.orderbySem);
		attr.addAll(stm.groupbySem);
		attr.addAll(stm.subRefered);

		for (Attribute a : qn.getSchema().getAllAttributes())
			if (attr.contains(a))
				pa.add(a);
		return new ProjectNode(qn, pa, tx);
	}
}
